package com.raymond.queue.client;

import com.raymond.queue.client.config.ThreadPoolConfig;
import com.raymond.queue.client.msg.MessageManages;
import com.raymond.queue.client.netty.ClientHandlersInitializer;
import com.raymond.queue.client.netty.ExponentialBackOffRetry;
import com.raymond.queue.client.netty.RetryPolicy;
import com.raymond.queue.client.queue.PackBlockingQueue;
import com.raymond.queue.client.service.Callback;
import com.raymond.queue.client.service.Consumption;
import com.raymond.queue.client.service.Production;
import com.raymond.queue.client.thread.AsyncCallBackThread;
import com.raymond.queue.client.thread.AsyncTaskHandleThread;
import com.raymond.queue.client.thread.PackRequestThread;
import com.raymond.queue.common.codec.BaseCodecStrategy;
import com.raymond.queue.common.msg.BaseMessage;
import com.raymond.queue.common.utils.CachedMillisecondClock;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 队列客户端
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-19 11:39
 */
public class QueueClient {
    private final static Logger logger = LoggerFactory.getLogger(QueueClient.class);
    /** ip **/
    private String host;
    /** port **/
    private int port;
    /** 响应超时时间 **/
    private int timeout = 15000;

    private final Bootstrap bootstrap = new Bootstrap();
    /**
     * 消息管理
     */
    private final ConcurrentHashMap<Integer, MessageManages> msgManager = new ConcurrentHashMap<>();

    private AsyncCallBackThread asyncCallBackThread;
    
    private final BlockingQueue<MessageManages> asyncTask = new LinkedBlockingQueue<>();

    private final PackBlockingQueue packTask = new PackBlockingQueue();

    private boolean isConnect = false;

    /** 重连策略 */
    private RetryPolicy retryPolicy;
    /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
    private Channel channel;

    private int packCount = 10000;

    public QueueClient(String host, int port) {
        this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
    }

    public QueueClient(String host, int port, RetryPolicy retryPolicy) {
        this.host = host;
        this.port = port;
        this.retryPolicy = retryPolicy;
        init();
    }



    /**
     * 向远程TCP服务器请求连接
     */
    public void connect() {
        synchronized (bootstrap) {
            ChannelFuture future = bootstrap.connect(host, port);
            future.addListener(getConnectionListener());
            this.channel = future.channel();
        }
    }

    public RetryPolicy getRetryPolicy() {
        return retryPolicy;
    }

    private void init() {
        try {
            BaseCodecStrategy.init("com.raymond.queue.client.codec");
        } catch (IllegalAccessException | InstantiationException e) {
            throw new RuntimeException("初始化编解码类异常", e);
        }
        asyncThreadInit();
        EventLoopGroup group = new NioEventLoopGroup();
        // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ClientHandlersInitializer(QueueClient.this));
    }

    private void asyncThreadInit() {
        clearAsyncTaskThreadInit();
        callBackInit();
        packThreadInit();
    }

    private void clearAsyncTaskThreadInit() {
        Thread clearAsyncTask = new AsyncTaskHandleThread(asyncTask, timeout, this);
        clearAsyncTask.start();
    }

    private void callBackInit() {
        ThreadPoolConfig threadPoolConfig = new ThreadPoolConfig("callBack");
        this.asyncCallBackThread = new AsyncCallBackThread(threadPoolConfig);
    }

    private void packThreadInit() {
        PackRequestThread packRequestThread = new PackRequestThread(this.packTask, asyncTask, this, packCount);
        packRequestThread.start();
    }

    private ChannelFutureListener getConnectionListener() {
        return future -> {
            if (!future.isSuccess()) {
                future.channel().pipeline().fireChannelInactive();
            } else {
                synchronized (this) {
                    this.isConnect = true;
                    notifyAll();
                }
            }
        };
    }

    public boolean isDone() {
        return isConnect;
    }

    public synchronized void sync() throws InterruptedException {
        if (isDone()) {
            return;
        }
        wait();
    }

    public boolean isActive() {
        return channel.isActive();
    }

    public ChannelFuture writeAndFlush(BaseMessage baseMessage) {
        return channel.writeAndFlush(baseMessage);
    }

    public MessageManages syncSend(BaseMessage req) {
        if (!channel.isWritable()) {
            logger.debug("缓冲区满了");
        }
        MessageManages messageManages = new MessageManages(req);
        msgManager.put(req.getSequenceId(), messageManages);
        channel.writeAndFlush(req);
        messageManages.wait0(timeout);
        if (messageManages.isDone()) {
            return messageManages;
        }
        msgManager.remove(req.getSequenceId());
        return messageManages.setCode(-2);
    }

    public void asyncSend(BaseMessage req, Callback callback) {
        asyncSend(new MessageManages(req, callback, asyncCallBackThread));
    }

    private void asyncSend(MessageManages messageManages) {
        msgManager.put(messageManages.getReq().getSequenceId(), messageManages);
        channel.writeAndFlush(messageManages.getReq()).addListener(future -> {
            if (future.isSuccess()) {
                asyncTask.add(messageManages.setStartTime(CachedMillisecondClock.INS.now()));
            } else if (messageManages.isRetry()) {
                asyncSend(messageManages);
            } else {
                messageManages.asyncSendFail();
            }
        });

    }

    public void asyncPackSend(BaseMessage req, Callback callback) {
        MessageManages messageManages = new MessageManages(req, callback, asyncCallBackThread);
        msgManager.put(req.getSequenceId(), messageManages);
        packTask.add(req);
        if (packTask.size() > packCount) {
            packTask.notify0();
            try {
                packTask.wait0(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
//        asyncTask.add(messageManages);
    }

    public Production getProduction() {
        return new Production(this);
    }

    public Consumption getConsumption() {
        return new Consumption(this);
    }

    public MessageManages remove(Integer sequenceId) {
        return msgManager.remove(sequenceId);
    }


    public ConcurrentHashMap<Integer, MessageManages> getMsgManager() {
        return this.msgManager;
    }


}
